/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.broker.processor;

import io.netty.channel.ChannelHandlerContext;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.broker.BrokerController;
import com.alibaba.rocketmq.broker.client.ClientChannelInfo;
import com.alibaba.rocketmq.broker.client.ConsumerGroupInfo;
import com.alibaba.rocketmq.broker.mqtrace.ConsumeMessageContext;
import com.alibaba.rocketmq.broker.mqtrace.ConsumeMessageHook;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.constant.PermName;
import com.alibaba.rocketmq.common.protocol.RequestCode;
import com.alibaba.rocketmq.common.protocol.ResponseCode;
import com.alibaba.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.GetConsumerListByGroupResponseBody;
import com.alibaba.rocketmq.common.protocol.header.GetConsumerListByGroupResponseHeader;
import com.alibaba.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.QueryConsumerOffsetResponseHeader;
import com.alibaba.rocketmq.common.protocol.header.UnregisterClientRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.UnregisterClientResponseHeader;
import com.alibaba.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.UpdateConsumerOffsetResponseHeader;
import com.alibaba.rocketmq.common.protocol.heartbeat.ConsumerData;
import com.alibaba.rocketmq.common.protocol.heartbeat.HeartbeatData;
import com.alibaba.rocketmq.common.protocol.heartbeat.ProducerData;
import com.alibaba.rocketmq.common.subscription.SubscriptionGroupConfig;
import com.alibaba.rocketmq.common.sysflag.TopicSysFlag;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
import com.alibaba.rocketmq.remoting.netty.NettyRequestProcessor;
import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;

/**
 * Client注册与注销管理
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-26
 */
public class ClientManageProcessor implements NettyRequestProcessor {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.BrokerLoggerName);

	private final BrokerController brokerController;

	public ClientManageProcessor(final BrokerController brokerController) {
		this.brokerController = brokerController;
	}

	@Override
	public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
		switch (request.getCode()) {
			case RequestCode.HEART_BEAT :
				// 心跳消息处理
				return this.heartBeat(ctx, request);
			case RequestCode.UNREGISTER_CLIENT :
				return this.unregisterClient(ctx, request);
			case RequestCode.GET_CONSUMER_LIST_BY_GROUP :
				// 获取consumerGroup名下的所有Consumer的ClientId
				return this.getConsumerListByGroup(ctx, request);
			case RequestCode.UPDATE_CONSUMER_OFFSET :
				// 更新消费进度Consumer Offset
				return this.updateConsumerOffset(ctx, request);
			case RequestCode.QUERY_CONSUMER_OFFSET :
				// 查询消费进度（QUERY_CONSUMER_OFFSET）
				return this.queryConsumerOffset(ctx, request);
			default :
				break;
		}
		return null;
	}

	/**
	 * 消费每条消息会回调
	 */
	private List<ConsumeMessageHook> consumeMessageHookList;

	public boolean hasConsumeMessageHook() {
		return consumeMessageHookList != null && !this.consumeMessageHookList.isEmpty();
	}

	public void registerConsumeMessageHook(List<ConsumeMessageHook> consumeMessageHookList) {
		this.consumeMessageHookList = consumeMessageHookList;
	}

	public void executeConsumeMessageHookAfter(final ConsumeMessageContext context) {
		if (hasConsumeMessageHook()) {
			for (ConsumeMessageHook hook : this.consumeMessageHookList) {
				try {
					hook.consumeMessageAfter(context);
				} catch (Throwable e) {
				}
			}
		}
	}

	private RemotingCommand updateConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
		final RemotingCommand response = RemotingCommand.createResponseCommand(UpdateConsumerOffsetResponseHeader.class);
		// final UpdateConsumerOffsetResponseHeader responseHeader =
		// (UpdateConsumerOffsetResponseHeader) response.readCustomHeader();
		final UpdateConsumerOffsetRequestHeader requestHeader = (UpdateConsumerOffsetRequestHeader) request.decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);

		// 消息轨迹：记录已经消费成功并提交 offset 的消息记录
		/**
		 * 检查ClientManageProcessor.consumeMessageHookList变量中是否设置了消费消息之后的回调类
		 * （在BrokerController.registerConsumeMessageHook(ConsumeMessageHook hook)方法中可以设置）
		 */
		if (this.hasConsumeMessageHook()) {
			// 执行hook
			// 若设置了则构建ConsumeMessageContext对象
			ConsumeMessageContext context = new ConsumeMessageContext();
			context.setConsumerGroup(requestHeader.getConsumerGroup());
			context.setTopic(requestHeader.getTopic());
			context.setClientHost(RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
			context.setSuccess(true);
			context.setStatus(ConsumeConcurrentlyStatus.CONSUME_SUCCESS.toString());
			final SocketAddress storeHost = new InetSocketAddress(brokerController.getBrokerConfig().getBrokerIP1(), brokerController.getNettyServerConfig().getListenPort());

			long preOffset = this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());
			Map<String, Long> messageIds = this.brokerController.getMessageStore().getMessageIds(requestHeader.getTopic(), requestHeader.getQueueId(), preOffset, requestHeader.getCommitOffset(), storeHost);
			context.setMessageIds(messageIds);
			// 执行ConsumeMessageHook.consumeMessageAfter(ConsumeMessageContext context)方法；
			this.executeConsumeMessageHookAfter(context);
		}
		/**
		 * 以请求消息中的topic和group构成"topic"@"consumerGroup"为key值，以queueId和请求消息中的offset构成Map存入ConsumerOffsetManager.offsetTable: ConcurrentHashMap<String==>topic@group , ConcurrentHashMap<Integer, Long>>变量中；
		 */
		this.brokerController.getConsumerOffsetManager().commitOffset(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
		response.setCode(ResponseCode.SUCCESS);
		response.setRemark(null);
		return response;
	}

	private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
		final RemotingCommand response = RemotingCommand.createResponseCommand(QueryConsumerOffsetResponseHeader.class);
		final QueryConsumerOffsetResponseHeader responseHeader = (QueryConsumerOffsetResponseHeader) response.readCustomHeader();
		final QueryConsumerOffsetRequestHeader requestHeader = (QueryConsumerOffsetRequestHeader) request.decodeCommandCustomHeader(QueryConsumerOffsetRequestHeader.class);

		/**
		 * 根据topic@consumerGroup、queueId从ConsumerOffsetManager.offsetTable变量中获取offset值并返回；若从offsetTable中查询不到，则返回-1；
		 */
		long offset = this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());

		// 订阅组存在
		if (offset >= 0) {
			responseHeader.setOffset(offset);
			response.setCode(ResponseCode.SUCCESS);
			response.setRemark(null);
		}
		// 订阅组不存在
		else {
			long minOffset = this.brokerController.getMessageStore().getMinOffsetInQuque(requestHeader.getTopic(), requestHeader.getQueueId());
			// 订阅组不存在情况下，如果这个队列的消息最小Offset是0，则表示这个Topic上线时间不长，服务器堆积的数据也不多，那么这个订阅组就从0开始消费。
			// 尤其对于Topic队列数动态扩容时，必须要从0开始消费。
			/**
			 * minOffset <= 0若第2步返回的最小逻辑offset值不大于零（可能未来得及创建该topic和queueId下面的consumequeue或者刚启动还未更新最小逻辑offset）
			 * !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(requestHeader.getTopic(), requestHeader.getQueueId(), 0)
			 * 检查该consumequeue队列的第一个消息单元中的物理偏移量与commitlog中最大物理偏移量的差值是否大于内存的最大使用率，若不大于内存的最大使用率，认为堆积的数据不多，则返回请求者消费进度为0，表示从0开始消费；否则返回未查询到结果；
			 */
			if (minOffset <= 0 && !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(requestHeader.getTopic(), requestHeader.getQueueId(), 0)) {
				responseHeader.setOffset(0L);
				response.setCode(ResponseCode.SUCCESS);
				response.setRemark(null);
			}
			// 新版本服务器不做消费进度纠正
			else {
				response.setCode(ResponseCode.QUERY_NOT_FOUND);
				response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first");
			}
		}

		return response;
	}

	/**
	 * 获取请求消息中consumerGroup下面的注册的所有Consumer的ClientId
	 * @param ctx
	 * @param request
	 * @return
	 * @throws RemotingCommandException
	 */
	public RemotingCommand getConsumerListByGroup(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
		final RemotingCommand response = RemotingCommand.createResponseCommand(GetConsumerListByGroupResponseHeader.class);
		final GetConsumerListByGroupRequestHeader requestHeader = (GetConsumerListByGroupRequestHeader) request.decodeCommandCustomHeader(GetConsumerListByGroupRequestHeader.class);

		// 根据请求消息中的consumerGroup从ConsumerManager.consumerTable:ConcurrentHashMap<String/* Group */, ConsumerGroupInfo>变量中获取对应的ConsumerGroupInfo对象；
		ConsumerGroupInfo consumerGroupInfo = this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup());
		if (consumerGroupInfo != null) {
			/**
			 * 若ConsumerGroupInfo对象不为null，
			 * 则从ConsumerGroupInfo. channelInfoTable:ConcurrentHashMap<Channel,ClientChannelInfo>变量中
			 * 获取每个ClientChannelInfo对象，取该对象的clientId变量，构成clientId的集合；
			 */
			List<String> clientIds = consumerGroupInfo.getAllClientId();
			if (!clientIds.isEmpty()) {
				GetConsumerListByGroupResponseBody body = new GetConsumerListByGroupResponseBody();
				body.setConsumerIdList(clientIds);
				response.setBody(body.encode());
				response.setCode(ResponseCode.SUCCESS);
				response.setRemark(null);
				return response;
			} else {
				log.warn("getAllClientId failed, {} {}", requestHeader.getConsumerGroup(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
			}
		} else {
			log.warn("getConsumerGroupInfo failed, {} {}", requestHeader.getConsumerGroup(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
		}

		response.setCode(ResponseCode.SYSTEM_ERROR);
		response.setRemark("no consumer for this group, " + requestHeader.getConsumerGroup());
		return response;
	}

	public RemotingCommand unregisterClient(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
		final RemotingCommand response = RemotingCommand.createResponseCommand(UnregisterClientResponseHeader.class);
		final UnregisterClientRequestHeader requestHeader = (UnregisterClientRequestHeader) request.decodeCommandCustomHeader(UnregisterClientRequestHeader.class);

		ClientChannelInfo clientChannelInfo = new ClientChannelInfo(//
				ctx.channel(), //
				requestHeader.getClientID(), //
				request.getLanguage(), //
				request.getVersion()//
		);

		// 注销Producer
		{
			final String group = requestHeader.getProducerGroup();
			if (group != null) {
				this.brokerController.getProducerManager().unregisterProducer(group, clientChannelInfo);
			}
		}

		// 注销Consumer
		{
			final String group = requestHeader.getConsumerGroup();
			if (group != null) {
				this.brokerController.getConsumerManager().unregisterConsumer(group, clientChannelInfo);
			}
		}

		response.setCode(ResponseCode.SUCCESS);
		response.setRemark(null);
		return response;
	}

	public RemotingCommand heartBeat(ChannelHandlerContext ctx, RemotingCommand request) {
		RemotingCommand response = RemotingCommand.createResponseCommand(null);

		HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);

		// 初始化ClientChannelInfo对象
		ClientChannelInfo clientChannelInfo = new ClientChannelInfo(//
				ctx.channel(), //
				heartbeatData.getClientID(), //
				request.getLanguage(), //
				request.getVersion()//
		);

		// 注册Consumer
		// 在接受到Consumer端的心跳信息后处理下列逻辑
		for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
			// 根据GroupName查找订阅组信息
			SubscriptionGroupConfig subscriptionGroupConfig = this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(data.getGroupName());
			if (null != subscriptionGroupConfig) {
				// 如果是单元化模式，则对 topic 进行设置
				int topicSysFlag = 0;
				if (data.isUnitMode()) {
					topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
				}
				// %RETRY%+GroupName
				String newTopic = MixAll.getRetryTopic(data.getGroupName());
				/**
				 * 创建以%RETRY%+GroupName为topic值的topic配置信息。
				 * 首先以该topic值在TopicConfigManager.topicConfigTable中查找是否存在，
				 * 若不存在则创建TopicConfig对象，并存入topicConfigTable中，同时将该topicConfigTable变量的值持久化到topics.json文件中；
				 */
				this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(//
						newTopic, //
						subscriptionGroupConfig.getRetryQueueNums(), //
						PermName.PERM_WRITE | PermName.PERM_READ, topicSysFlag);
			}

			// 注册Consumer
			// 就是更新ConsumerManager.consumerTable:ConcurrentHashMap<String/* Group */, ConsumerGroupInfo>变量的值
			boolean changed = this.brokerController.getConsumerManager().registerConsumer(//
					data.getGroupName(), //
					clientChannelInfo, //
					data.getConsumeType(), //
					data.getMessageModel(), //
					data.getConsumeFromWhere(), //
					data.getSubscriptionDataSet()//
			);

			if (changed) {
				log.info("registerConsumer info changed {} {}", //
						data.toString(), //
						RemotingHelper.parseChannelRemoteAddr(ctx.channel())//
				);

				// TODO:有可能会有频繁变更
				// for (SubscriptionData subscriptionData :
				// data.getSubscriptionDataSet()) {
				// this.brokerController.getTopicConfigManager().updateTopicUnitSubFlag(
				// subscriptionData.getTopic(), data.isUnitMode());
				// }
			}
		}

		// 注册Producer 就是更新ProducerManager.groupChannelTable:HashMap<String,HashMap<Channel,ClientChannelInfo>>变量值
		// 在接受到Producer的心跳信息后处理该逻辑。
		for (ProducerData data : heartbeatData.getProducerDataSet()) {
			this.brokerController.getProducerManager().registerProducer(data.getGroupName(), clientChannelInfo);
		}

		response.setCode(ResponseCode.SUCCESS);
		response.setRemark(null);
		return response;
	}
}
